// Connector.cpp : implementation file
//

#include "stdafx.h"
#include "DataReceiver.h"
#include "DataSClient.h"

#ifdef _DEBUG
#undef THIS_FILE
static char THIS_FILE[]=__FILE__;
#define new DEBUG_NEW
#endif

/////////////////////////////////////////////////////////////////////////////
// CDPSocket

CNetSocket::CNetSocket()
{
	pDataReceiver=NULL;
}

CNetSocket::CNetSocket(CDataReceiver *pdatareceiver)
{
	pDataReceiver=pdatareceiver;
}

CNetSocket::~CNetSocket()
{

}

/////////////////////////////////////////////////////////////////////////////
// CDPSocket member functions

void CNetSocket::OnClose(int nErrorCode) 
{
	ASSERT(pDataReceiver!=NULL);

	Status=TH_OFF;
	pDataReceiver->NetClose(this);
	CSocket::OnClose(nErrorCode);
}

void CNetSocket::OnReceive(int nErrorCode) 
{
	ret=TRUE;
	nRead=Receive(packet, MAX_NET_DATAPACKET);

	if(nRead>=MIN_NET_PACKETSIZE)
	{
		cur_pos=0;
		ptr=packet;
		while(cur_pos<=nRead-MIN_NET_PACKETSIZE && ret)
		{
			data_size=*((int *)(ptr+2));
			type=ptr[0];
			
			if(cur_pos+data_size<=nRead && data_size>=MIN_NET_PACKETSIZE)
			{
				buff=new char[data_size];

				for(i=0;i<data_size;i++) buff[i]=*ptr++;

				// todo check if data size equal to expected
				if(type==NET_TYPE_TICK && data_size>pDataReceiver->tick_size+NET_DATA_HEADER) pDataReceiver->PostThreadMessage(WM_OnNewTickArrived,(WPARAM)buff,(LPARAM)data_size);
				else if(type==NET_TYPE_BIDASK && data_size>pDataReceiver->bidask_size+NET_DATA_HEADER) pDataReceiver->PostThreadMessage(WM_OnNewBidAskArrived,(WPARAM)buff,(LPARAM)data_size);
				else if(type==NET_TYPE_INTEREST && data_size>pDataReceiver->common_size+NET_DATA_HEADER) pDataReceiver->PostThreadMessage(WM_OnNewCommonArrived,(WPARAM)buff,(LPARAM)data_size);
				else if(type==NET_TYPE_INFO) pDataReceiver->PostThreadMessage(WM_OnNewMessageReceived,(WPARAM)buff,(LPARAM)data_size);
				else 
				{
					delete [] buff;
					pDataReceiver->Log("Invalid data received from socket.",IDM_ERROR,254);
					ret=FALSE;
				}

				cur_pos+=data_size;
			}
			else ret=FALSE;
		}
	}
	else if(nRead==SOCKET_ERROR) 
	{
		nRead=GetLastError();
		pDataReceiver->Log("Error while reading data from socket.",IDM_ERROR,255);
		if(nRead!=WSAEWOULDBLOCK && nRead!=WSAEMSGSIZE) Close();
	}
	else if(nRead==0) Close();

	CSocket::OnReceive(nErrorCode);
}


/////////////////////////////////////////////////////////////////////////////
// CConnector

IMPLEMENT_DYNCREATE(CDataReceiver, CWinThread)

CDataReceiver::CDataReceiver()
{
	Settings=theApp.SettingsDataReceiver;
	//Verbosity=theApp.Settings.verbosity;
	Status=TH_OFF;

	hTimer=NULL;
	hTimerQueue=NULL;
	Disconnecting=FALSE;
	Connecting=FALSE;
	ActiveCount=0;
	ModuleActionDone=NULL;
	NetSockets=NULL;
	ReceivedTicks=0;
	ReceivedBidAsks=0;
	ReceivedCommon=0;
	tick_size=sizeof(STick);
	bidask_size=36;//sizeof(SBidAsk)-sizeof(double*)*2-sizeof(int*)*2;
	common_size=sizeof(SCommon);

	for(int i=0;i<PL_COUNT;i++) data_pending[i]=FALSE;

	Settings.allow_ticks = true;
	Settings.allow_bidask = true;
}

CDataReceiver::~CDataReceiver()
{

}

void CDataReceiver::Log(LPCTSTR message, BYTE type, int code)
{
	if(message==NULL || !(1<<(type-1) & Verbosity)) return;

	//SLogEvent* LogEvent=new SLogEvent;
	//LogEvent->code=code;
	//LogEvent->time=theApp.CurrentTime;
	//LogEvent->module=ID_DATARECEIVER;
	////LogEvent->module_id="";
	////LogEvent->ticker_name="";
	//LogEvent->type=type;
	//LogEvent->message=message;

	//if(theApp.ControlServer) theApp.ControlServer->PostThreadMessage(WM_OnLogMessage, (WPARAM)LogEvent, 0);
}

BOOL CDataReceiver::IsRunning()
{
	if(Status==TH_ON) return TRUE;
	else return FALSE;
}

BOOL CDataReceiver::InitInstance()
{
	CString str;

	Status=TH_PAUSED;
	
	str.Format("%d datareceiver",theApp.ProcID);
	ModuleActionDone=CreateEvent(NULL,TRUE,FALSE,str);

	hTimerQueue=CreateTimerQueue();
	if(hTimerQueue==NULL) Log("Failed to create queue for timer.",IDM_ERROR,10);

	if(hTimerQueue==NULL || !InitMessageLoop())
	{
		Status=TH_ERROR;
		SetEvent(theApp.InitDone);
		return TRUE;
	}
	
	if (!AfxSocketInit()) 
	{
		Log("Windows sockets initialization failed.",IDM_ERROR,21);
		Status=TH_ERROR;
	}
	else
	{
		//NetSockets=new CNetSocket;
		NetSockets=NULL;
	}



	SetEvent(theApp.InitDone);
	
	return TRUE;
}

int CDataReceiver::ExitInstance()
{
	Status=TH_EXIT;


	if(ActiveCount>0) 
	{
		OnDisconnect();
		ProcessMessage(OPER_SLEEP_MS*3);
	}
	
	if(NetSockets)
	{ 	
		delete NetSockets;
	}


	CloseHandle(ModuleActionDone);

	DeInitMessageLoop();

	SetEvent(theApp.InitDone);

	this->SuspendThread();
	return CWinThread::ExitInstance();
}

void CDataReceiver::ProcessMessage(int timeout_ms)
{
	CMySTime starttime;
	MSG m_msgCur;

	starttime=theApp.CurrentTime;

	// processing cycle
	while(1)
	{		
		while(PeekMessage(&m_msgCur, (HWND)NULL, NULL, NULL, PM_REMOVE)) 
			DispatchMessage(&m_msgCur);

		if(timeout_ms<=0 || theApp.CurrentTime-starttime>timeout_ms) return;
	}
}

void CDataReceiver::NetClose(CNetSocket *pSocket) 
{

	delete pSocket;

	if(Disconnecting) Log("Connection with DP's server closed by remote host.");
	ResetEvent(ModuleActionDone);
}

BOOL CDataReceiver::StartTimer(int Ms)
{
	if(hTimer) StopTimer();

	CreateTimerQueueTimer(&hTimer, hTimerQueue, (WAITORTIMERCALLBACK)TimerCallback, (PVOID)this, Ms, 0, WT_EXECUTEINTIMERTHREAD);
	if(hTimer==NULL) 
	{
		Log("Couldn't set timer. Connection delay functionality will not work.",IDM_ERROR,13);
		return FALSE;
	}
	return TRUE;
}

void CDataReceiver::StopTimer()
{
	DeleteTimerQueueTimer(hTimerQueue,hTimer,INVALID_HANDLE_VALUE);
	hTimer=NULL;
}

VOID CALLBACK CDataReceiver::TimerCallback(PVOID lpParam, BOOLEAN TimerOrWaitFired)
{
	if(lpParam) ((CDataReceiver*)lpParam)->PostThreadMessage(WM_TIMER, 0, 0);
}

BOOL CDataReceiver::InitMessageLoop()
{
	CString str;
	BOOL Failed=FALSE;

	NewMessage=NULL;
	LastMessagePtr=NULL;
	FirstMessagePtr=NULL;
	CurMessagePtr=NULL;
	MessageLoopSize=0;
	int i;
	str.Format("%d datareceiver loop",theApp.ProcID);
	NewMessage=CreateEvent(NULL,FALSE,FALSE,str);

	if(NewMessage!=NULL && GetLastError()!=ERROR_ALREADY_EXISTS)
	{
		if(InitializeCriticalSectionAndSpinCount(&LoopLock,4000))
		{
			EnterCriticalSection(&LoopLock);

			MessageLoopSize=EVENT_LOOP_SIZE_HUGE;
			FirstMessagePtr=new SMessage[MessageLoopSize];
			for(i=0; i< MessageLoopSize; i++)
			{
				(FirstMessagePtr + i)->Status = 0;
			}
			if(FirstMessagePtr)
			{
				LastMessagePtr=FirstMessagePtr+MessageLoopSize-1;
				LastMessagePtr->Status=2;
				FirstMessagePtr->Status = 0;
				LastMessagePtr=FirstMessagePtr;
				CurMessagePtr=FirstMessagePtr;
			}
			else Failed=TRUE;

			LeaveCriticalSection(&LoopLock);
		}
		else Failed=TRUE;
	}
	else 
	{
		NewMessage=NULL;
		Failed=TRUE;
	}

	if(Failed) 
	{
		DeInitMessageLoop();
		Log("Failed to initialize message loop.",IDM_ERROR,15);
	}

	return !Failed;
}

void CDataReceiver::DeInitMessageLoop()
{
	BOOL Initialized;

	if(MessageLoopSize) Initialized=TRUE;
	else Initialized=FALSE;

	if(Initialized) EnterCriticalSection(&LoopLock);

	if(NewMessage) CloseHandle(NewMessage);
	if(FirstMessagePtr)
	{
		delete [] FirstMessagePtr;
	}

	NewMessage=NULL;
	FirstMessagePtr=NULL;
	LastMessagePtr=NULL;
	CurMessagePtr=NULL;
	MessageLoopSize=0;

	if(Initialized) DeleteCriticalSection(&LoopLock);
}

BOOL CDataReceiver::PostThreadMessage(UINT message, WPARAM wParam, LPARAM lParam)
{
	if(MessageLoopSize) EnterCriticalSection(&LoopLock);
	if(!MessageLoopSize) return FALSE;

	if(CurMessagePtr->Status!=0) // waiting for empty row
	{
		Log("Message buffer overflow detected.",IDM_ALERT,2001);

		for(int i=0;i<50 && CurMessagePtr->Status!=0;i++) Sleep(OPER_SLEEP_MS);
	}

	if(CurMessagePtr->Status==0)
	{
		CurMessagePtr->Message=message;
		CurMessagePtr->wParam=wParam;
		CurMessagePtr->lParam=lParam;
		CurMessagePtr->Status=1;

		CurMessagePtr++;
		if(CurMessagePtr->Status==2) CurMessagePtr=FirstMessagePtr;

		SetEvent(NewMessage);
		LeaveCriticalSection(&LoopLock);

		return TRUE;
	}
	else 
	{
		LeaveCriticalSection(&LoopLock);
		return FALSE;
	}
}

int CDataReceiver::Run()
{
	TRACE("START Run()\n");
	BOOL NewMessageExist=FALSE;
	BOOL finalized=FALSE;
	UINT message;
	MSG m_msgCur;

	PostThreadMessage(WM_OnConnect,(WPARAM)NULL,(LPARAM)NULL);
	// main processing cycle
	while(1)
	{		
		while(PeekMessage(&m_msgCur, (HWND)NULL, NULL, NULL, PM_REMOVE)) 
			DispatchMessage(&m_msgCur);

		if(NewMessageExist || WaitForSingleObject(NewMessage,0)==WAIT_OBJECT_0)
		{
			NewMessageExist=FALSE;
			TRACE("Run: New message\n");
			while(LastMessagePtr->Status==1) 
			{
				message=LastMessagePtr->Message;

				if(message==WM_OnNewTickArrived) 
				{
					buff=(char *)LastMessagePtr->wParam;
					OnNewTickArrived(buff);
					delete [] buff;
				}
				else if(message==WM_OnNewBidAskArrived) 
				{
					buff=(char *)LastMessagePtr->wParam;
					OnNewBidAskArrived(buff);
					delete [] buff;
				}
				else if(message==WM_OnNewCommonArrived) 
				{
					buff=(char *)LastMessagePtr->wParam;
					OnNewCommonArrived(buff);
					delete [] buff;
				}
				else if(message==WM_OnNewMessageReceived) 
				{
					buff=(char *)LastMessagePtr->wParam;
					OnNewMessageReceived(buff,(int)LastMessagePtr->lParam);
					delete [] buff;
				}
				else if(message==WM_TIMER) OnTimer(LastMessagePtr->wParam, LastMessagePtr->lParam);
				else if(message==WM_OnSuspend) OnSuspend();
				else if(message==WM_OnResume) OnResume();
				else if(message==WM_OnConnect) 
				{
					Connecting=TRUE;
					Disconnecting=FALSE;
					if((int)LastMessagePtr->wParam<=0 || !StartTimer((int)LastMessagePtr->wParam)) OnConnect();
				}
				else if(message==WM_OnDisconnect) 
				{
					Connecting=FALSE;
					Disconnecting=TRUE;
					if(hTimer) StopTimer();
					OnDisconnect();
				}
				else if(message==WM_QUIT) 
				{	
					return ExitInstance();
				}

				LastMessagePtr->Message=0;
				LastMessagePtr->lParam=0;
				LastMessagePtr->wParam=0;
				LastMessagePtr->Status=0;
				LastMessagePtr++;
				if(LastMessagePtr->Status==2) LastMessagePtr=FirstMessagePtr;
			} 
		}
		
		if(!finalized)
		{
			if(WaitForSingleObject(NewMessage,RUN_SLEEP_MS)==WAIT_OBJECT_0) NewMessageExist=TRUE;
		}
		else finalized=FALSE;
	}
	TRACE("EXIT Run\n");
}

// CDataReceiver message handlers

void CDataReceiver::OnTimer(WPARAM nTimerID,LPARAM lParam)
{
	if(!IsRunning()) return;

	if(Connecting && !Disconnecting) OnConnect();
	else if(Disconnecting && !Connecting) OnDisconnect();
}

void CDataReceiver::OnSuspend()
{
	if(IsRunning()) 
	{
		Status=TH_PAUSED;
	}

	SetEvent(ModuleActionDone);
}

void CDataReceiver::OnResume()
{
	if(!IsRunning() && Status!=TH_ERROR && Status!=TH_OFF)
	{
		Status=TH_ON;
	}

	SetEvent(ModuleActionDone);
}

void CDataReceiver::OnConnect()
{
	CString str;
	int size;
	int n;
	BOOL opt=TRUE;
	char *data, *ptr;

			if(NetSockets==NULL)
			{
				NetSockets=new CNetSocket(this);
				NetSockets->Status=TH_OFF;

				if(NetSockets->Create()==0)
				{
					delete NetSockets;
					NetSockets=NULL;

					Status=TH_ERROR;
					Log("Connection to DP's servers failed due to error: Couldn't create sockets.",IDM_ERROR,41);
					Connecting=FALSE;
					return;
				}
				else 
				{
					NetSockets->Status=TH_PAUSED;
					NetSockets->SetSockOpt(TCP_NODELAY,&opt,sizeof(BOOL),IPPROTO_TCP);
					size=262144;
					NetSockets->SetSockOpt(SO_RCVBUF,&size,sizeof(int),SOL_SOCKET);
				}
			}

		str.Format("%s;%d",IDS_FUll_APP_NAME,theApp.ProcID);

		size=(int)str.GetLength()+NET_DATA_HEADER;
		data=new char[size];
		data[0]=NET_TYPE_LOGIN;
		data[1]=1;
		ptr=(char *)(&size);
		for(n=0;n<4;n++) data[n+2]=ptr[n];
		for(n=NET_DATA_HEADER;n<size;n++) data[n]=str[n-NET_DATA_HEADER];

			if(NetSockets->Status!=TH_ON)
			{
				if(NetSockets->Connect(theApp.m_ip.GetString(),theApp.m_port))
				{
					if(NetSockets->Send(data,size)==size) 
					{
						NetSockets->Status=TH_ON;
						ActiveCount++;
					}
					else 
					{
						NetSockets->Close();
						NetSockets->Status=TH_PAUSED;
					}
				}
			}

		delete [] data;
	Status = NetSockets->Status;
	SetEvent(ModuleActionDone);
}

void CDataReceiver::OnDisconnect()
{
	int disc=0;
	CString str;

	if(!IsRunning()) return;

			if(NetSockets!=NULL)
			{
				if(NetSockets->Status==TH_ON) 
				{
					NetSockets->Close();
					ActiveCount--;
					disc++;
				}
	
				delete NetSockets;
				NetSockets=NULL;
			}

	Disconnecting=FALSE;
	Status = TH_ERROR;
	SetEvent(ModuleActionDone);
}

void CDataReceiver::OnNewTickArrived(char *buffer)
{
	if(!IsRunning() || !Settings.allow_ticks) return;

	if(buffer[1]!=1) return;

	//if(buffer[NET_DATA_HEADER]==0 || !theApp.TickersCodeStr.Lookup(buffer+NET_DATA_HEADER,(void*&)ticker_info)) return;

	ReceivedTicks++;
	cur=NET_DATA_HEADER+1;

	while(buffer[cur++]!=0);
	
	tick=new STick;
	target=(char *)tick;
	for(n=0;n<tick_size;n++) target[n]=buffer[cur++];

	platform=tick->platform;
	delay=theApp.CurrentTime-tick->time;
	tick->delay=delay;
	tick->flags=tick->flags | IDD_REMOTE_DATA;

	theApp.m_pMainWnd->PostMessageA(WM_OnNewTickArrived,(WPARAM)tick,(LPARAM)ticker_info);

	if(!data_pending[platform])
	{
		data_pending[platform]=TRUE;
		data_time[platform]=theApp.CurrentTime;
	}
}

void CDataReceiver::OnNewBidAskArrived(char *buffer)
{
	if(!IsRunning() || !Settings.allow_bidask) return;

	if(buffer[1]!=1) return;

	//if(buffer[NET_DATA_HEADER]==0 || !theApp.TickersCodeStr.Lookup(buffer+NET_DATA_HEADER,(void*&)ticker_info)) return;

	ReceivedBidAsks++;
	cur=NET_DATA_HEADER+1;

	while(buffer[cur++]!=0);

	bidask=new SBidAsk;
	target=(char *)bidask;
	for(n=0;n<bidask_size;n++) target[n]=buffer[cur++];

	if(bidask->askrows>0)
	{
		bidask->ask=bidask->askrows;
		bidask->asksize=bidask->askrows;

		target=(char *)&bidask->ask;
		len=bidask->askrows*sizeof(double);
		for(n=0;n<len;n++) target[n]=buffer[cur++];

		target=(char *)&bidask->asksize;
		len=bidask->askrows*sizeof(int);
		for(n=0;n<len;n++) target[n]=buffer[cur++];
	}

	if(bidask->bidrows>0)
	{
		bidask->bid=bidask->bidrows;
		bidask->bidsize=bidask->bidrows;

		target=(char *)&bidask->bid;
		len=bidask->bidrows*sizeof(double);
		for(n=0;n<len;n++) target[n]=buffer[cur++];

		target=(char *)&bidask->bidsize;
		len=bidask->bidrows*sizeof(int);
		for(n=0;n<len;n++) target[n]=buffer[cur++];
	}

	platform=bidask->platform;
	bidask->flags=bidask->flags | IDD_REMOTE_DATA;

	if(bidask->ticknum==0) 
	{
		//bidask->index=++ticker_info->last_bidask.index;
		//bidask->ticknum=ticker_info->last_bidask.ticknum;
	}
	
	theApp.m_pMainWnd->PostMessageA(WM_OnNewBidAskArrived,(WPARAM)bidask,(LPARAM)ticker_info);

	if(!data_pending[platform])
	{
		data_pending[platform]=TRUE;
		data_time[platform]=theApp.CurrentTime;
	}
}

void CDataReceiver::OnNewCommonArrived(char *buffer)
{
	if(!IsRunning() || !Settings.allow_common) return;

	if(buffer[1]!=1) return;

	//if(buffer[NET_DATA_HEADER]==0 || !theApp.TickersCodeStr.Lookup(buffer+NET_DATA_HEADER,(void*&)ticker_info)) return;

	ReceivedCommon++;
	cur=NET_DATA_HEADER+1;

	while(buffer[cur++]!=0);

	common=new SCommon;
	target=(char *)common;
	for(n=0;n<common_size;n++) target[n]=buffer[cur++];

	platform=common->platform;
	common->flags=common->flags | IDD_REMOTE_DATA;

	//theApp.DataManager->PostThreadMessage(WM_OnNewCommonArrived,(WPARAM)common,(LPARAM)ticker_info);

	if(!data_pending[platform])
	{
		data_pending[platform]=TRUE;
		data_time[platform]=theApp.CurrentTime;
	}
}

void CDataReceiver::OnNewMessageReceived(char *message, int length)
{
	if(message) delete [] message;
}